package com.inspinia.base.mq.rabbitmq.service;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.inspinia.base.mq.rabbitmq.config.RabbitConfig;
import com.inspinia.base.mq.rabbitmq.config.RabbitMqConfig;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author Veblen
 */
@Component
@Slf4j
public class RabbitMqConsumerRegister {

    private Connection connection = RabbitMqConfig.getConnection();

    private ExecutorService es = newFixedThreadPoolWithQueueSize(32, 1024);

    /**
     * 添加正常的一个消费者
     * @param taskName
     * @param handler
     * @param isAutoAck
     * @param qos
     */
    public void addNormalCounsumer(String taskName, MessageHandler handler, boolean isAutoAck, int qos) {
        try {
            Channel channel = connection.createChannel();
            channel.queueDeclare(taskName, true, false, false, null);

            channel.basicQos(qos);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    try {
                        handler.handle(this.getChannel(), message, envelope.getDeliveryTag(), isAutoAck);
                    } catch (RejectedExecutionException e) {
                        if (!isAutoAck) {
                            channel.basicReject(envelope.getDeliveryTag(), true);
                        }
                        log.info("Mq message handle queue full");
                    } catch (Exception e) {
                        log.error("mq consume message error", e);
                    }
                }

                @Override
                public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
                    log.error("consumer:" + consumerTag + "被关闭", sig);
                }
            };
            channel.basicConsume(taskName, isAutoAck, consumer);
        } catch (Exception e) {
            log.error("mq consumer registe error", e);
        } finally {

        }
    }

    /**
     *添加一个延迟消费者
     * @param doneQueueName
     * @param routingkey
     * @param handler
     * @param autoAck
     * @param qos
     */
    public void addDelayConsumer(String doneQueueName, String routingkey, MessageHandler handler, boolean autoAck,
                                 int qos) {
        try {
            // 声明延迟队列的exchange
            Channel channel = connection.createChannel();
            Map<String, Object> args = new HashMap<>(1);
            args.put("x-delayed-type", "direct");
            channel.exchangeDeclare(RabbitConfig.DELAY_QUEUE_EXCHANGE, "x-delayed-message", true, false, args);
            // 声明队列
            channel.queueDeclare(doneQueueName, true, false, false, null);
            // 绑定队列
            channel.queueBind(doneQueueName, RabbitConfig.DELAY_QUEUE_EXCHANGE, doneQueueName);
            channel.basicQos(qos);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    try {
                        handler.handle(this.getChannel(), message, envelope.getDeliveryTag(), autoAck);
                    } catch (RejectedExecutionException e) {
                        if (!autoAck) {
                            // 只有没有开启ack时，才能重新入队列
                            channel.basicReject(envelope.getDeliveryTag(), true);
                        }
                        log.info("Mq delay message handle queue full");
                    } catch (Exception e) {
                        log.error("mq consumer message error", e);
                    }
                }

                @Override
                public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
                    log.error("consumer:" + consumerTag + "被关闭", sig);
                }

            };
            channel.basicConsume(doneQueueName, autoAck, consumer);
        } catch (Exception e) {
            log.error("mq consumer register error", e);
        }
    }

    /**
     * 队列满了后
     *
     * @param task
     */
    public void submitTask(Runnable task) throws InterruptedException {
        RejectedExecutionException ex = null;
        int len = 2;
        for (int i = 0; i < len; i++) {
            try {
                es.submit(task);
                return;
            } catch (RejectedExecutionException e) {
                ex = e;
                Thread.sleep(50);
            }
        }
        throw ex;
    }

    /**
     * 线程池对象
     * @param poolSize
     * @param queueLen
     * @return
     */
    public static ExecutorService newFixedThreadPoolWithQueueSize(int poolSize, int queueLen) {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("rabbitmqRegister-pool-%d").build();

        ExecutorService singleThreadPool = new ThreadPoolExecutor(poolSize, poolSize, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(queueLen),namedThreadFactory);
        return singleThreadPool;
    }

    /**
     * 消息重新排队
     * @param channel
     * @param deliveryTag
     */
    public void requeue(Channel channel, long deliveryTag) {
        try {
            channel.basicReject(deliveryTag, true);
        } catch (IOException e1) {
            log.error("消息重新排队失败", e1);
        }
    }

    /**
     * 消息处理接口
     */
    public interface MessageHandler {
        /**
         * 消息事件
         * @param channel
         * @param message
         * @param deliveryTag
         * @param autoAck
         * @throws InterruptedException
         */
        void handle(Channel channel, String message, long deliveryTag, boolean autoAck) throws InterruptedException;
    }

}